package cn.itcast.sql;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.sql
 * @date 2021/3/6 0006 15:06
 * @Copyright © 2015-04-29  One for each, and two for each
 *     演示flink tablesql 将datastream数据转为table或view 然后使用sql进行统计
 */
public class Demo2 {
    public static void main(String[] args) throws Exception {
        //TODO 0 env
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final EnvironmentSettings build = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env, build);
        //TODO 1 source
        final DataStreamSource<WC> wordsDS = env.fromElements(new WC("Hello", 1),
                new WC("World", 1),
                new WC("Hello", 1));
        //TODO 2 thransformation
        //将DataStream转为View或Table
        tenv.createTemporaryView("t_words", wordsDS,$("word"), $("frequency"));
       String sql ="select word,sum(frequency) as frequency\n "+
               "from t_words\n " +
               "group by word";
        //执行sql
        Table resultTable = tenv.sqlQuery(sql);
        //转为DataStream
        final DataStream<Tuple2<Boolean, WC>> resultDS = tenv.toRetractStream(resultTable, WC.class);
        //toAppendStream → 将计算后的数据append到结果DataStream中去
        //toRetractStream  → 将计算后的新的数据在DataStream原数据的基础上更新true或是删除false
        //类似StructuredStreaming中的append/update/complete
        //TODO 3 sink
        resultDS.print();
        //TODO 4 execute
        env.execute();
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WC {
        public String word;
        public long frequency;
    }
}
